package com.ytktt.study.week9.task3.rpcfx.server.netty;

import com.ytktt.study.week9.task3.rpcfx.api.RpcfxRequest;
import com.ytktt.study.week9.task3.rpcfx.api.RpcfxResponse;
import com.ytktt.study.week9.task3.rpcfx.server.RpcfxInvoker;
import com.ytktt.study.week9.task3.rpcfx.server.RpcfxServer;
import com.ytktt.study.week9.task3.rpcfx.server.netty.coder.FastJsonDecoder;
import com.ytktt.study.week9.task3.rpcfx.server.netty.coder.FastJsonEncoder;
import com.ytktt.study.week9.task3.rpcfx.server.netty.coder.ProtostuffDecoder;
import com.ytktt.study.week9.task3.rpcfx.server.netty.coder.ProtostuffEncoder;
import com.ytktt.study.week9.task3.rpcfx.server.netty.handler.NettyServerHandler;
import com.ytktt.study.week9.task3.rpcfx.server.properties.RpcfxServerProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpContentDecompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

/**
 * Netty Server
 *
 * @author ytk
 */
@Slf4j
public class NettyServer extends RpcfxServer {

    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;

    public NettyServer(RpcfxServerProperties rpcfxServerProperties, RpcfxInvoker rpcfxInvoker) {
        super(rpcfxServerProperties, rpcfxInvoker);
        RpcfxServerProperties.InBoundProperties inBound = rpcfxServerProperties.getInBound();
        this.bossGroup = new NioEventLoopGroup(inBound.getBossCount());
        this.workerGroup = new NioEventLoopGroup(inBound.getWorkerCount());
    }

    @Override
    public void start(RpcfxServerProperties rpcfxServerProperties, RpcfxInvoker rpcfxInvoker) throws Exception {
        RpcfxServerProperties.InBoundProperties inBound = rpcfxServerProperties.getInBound();
        ServerBootstrap b = new ServerBootstrap();
        b.option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.TCP_NODELAY, inBound.isTcpNoDelay())
                .childOption(ChannelOption.SO_KEEPALIVE, inBound.isKeepalived())
                .childOption(ChannelOption.SO_REUSEADDR, inBound.isReuseAddr())
                .childOption(ChannelOption.SO_RCVBUF, inBound.getRcvBuf() * 1024)
                .childOption(ChannelOption.SO_SNDBUF, inBound.getBossCount() * 1024)
                .childOption(EpollChannelOption.SO_REUSEPORT, inBound.isReusePort())
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

        b.group(this.bossGroup, this.workerGroup)
                .channel(NioServerSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.DEBUG))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new ProtostuffEncoder(RpcfxResponse.class));
                        p.addLast(new ProtostuffDecoder(RpcfxRequest.class));
                        p.addLast(new LoggingHandler(LogLevel.INFO));
                        p.addLast(new NettyServerHandler(rpcfxInvoker));
                    }
                });

        Channel ch = b.bind(rpcfxServerProperties.getPort()).sync().channel();
        log.info("开启netty tcp服务器，监听地址和端口为 127.0.0.1:{}/", rpcfxServerProperties.getPort());
        ch.closeFuture().sync();
    }

    @Override
    public void clear() {
        this.bossGroup.shutdownGracefully();
        this.workerGroup.shutdownGracefully();
    }
}
